//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// seq_scan_executor.cpp
//
// Identification: src/execution/seq_scan_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "execution/executors/seq_scan_executor.h"
#include <memory>
#include <utility>
#include "common/macros.h"
#include "concurrency/transaction.h"
#include "storage/table/table_iterator.h"

namespace bustub {

SeqScanExecutor::SeqScanExecutor(ExecutorContext *exec_ctx, const SeqScanPlanNode *plan) : AbstractExecutor(exec_ctx) {
  // table_iterator_ =
  // std::make_unique<TableIterator>(exec_ctx->GetCatalog()->GetTable(plan->table_oid_)->table_->MakeIterator());
  plan_ = plan;
  catalog_ = exec_ctx->GetCatalog();
}

void SeqScanExecutor::Init() {
  table_iterator_ = std::make_unique<TableIterator>(catalog_->GetTable(plan_->table_oid_)->table_->MakeIterator());
}

auto SeqScanExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (table_iterator_->IsEnd()) {
    return EXECUTOR_EXHAUSTED;
  }
  auto heap_tuple{table_iterator_->GetTuple()};
  while (heap_tuple.first.is_deleted_) {
    ++(*table_iterator_);
    if (table_iterator_->IsEnd()) {
      return EXECUTOR_EXHAUSTED;
    }
    heap_tuple = table_iterator_->GetTuple();
  }
  *tuple = std::move(heap_tuple.second);
  *rid = table_iterator_->GetRID();

  ++(*table_iterator_);
  return EXECUTOR_ACTIVE;
}

}  // namespace bustub
